package pipe;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.Pipe;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.Iterator;

public class Consumer implements Runnable{

    private Selector selector;

    public Consumer(Selector selector) {
        this.selector = selector;
    }

    @Override
    public void run() {
        System.out.println("consumer start");
        int round = 2;
        SelectionKey key;
        while (!Thread.currentThread().isInterrupted()) {
            try {
                int select = selector.select();
                if (select == 0){
                    System.out.println("空转一圈");
                    continue;
                }
                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                System.out.println(selector.selectedKeys());
                ByteBuffer buffer = ByteBuffer.allocate(16);
                while (iterator.hasNext()) {
                    key = iterator.next();
                    Pipe.SourceChannel channel =(Pipe.SourceChannel) key.channel();
                    System.out.println(Thread.currentThread().getName()+" : "+channel.validOps());
                    for (int i = 0; i < round; i++) {
                        int cl = channel.read(buffer);
                        if (cl==0) break;
                        if (cl==-1) {
                            key.cancel();
                            break;
                        }
                        buffer.flip();
                        byte[] bytes = new byte[buffer.remaining()];
                        buffer.get(bytes);
                        System.out.println(Thread.currentThread().getName()+" :" +new String(bytes));

                        buffer.clear();
                    }

                    iterator.remove();
                }

            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
